package com.winit.core.mq.broker;

import com.winit.common.util.DateUtil;
import com.winit.common.util.ObjectToString;
import com.winit.core.mq.listener.QueueListener;
import com.winit.core.mq.message.Message;
import com.winit.core.mq.queue.MessageQueue;
import com.winit.core.mq.queue.impl.AsyncMessageQueue;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by xiangyu.liang on 2016/8/15.
 */
@Component
public class MessageBroker {
    private Map<String,MessageQueue> queueMap;
    public MessageBroker(){
        queueMap=new ConcurrentHashMap<>();
    }
    public void sendMessage(String queueName,Message message){
        if(!queueMap.keySet().contains(queueName)){
            return;
        }
        MessageQueue messageQueue=queueMap.get(queueName);
        System.out.println(DateUtil.getCurDateStr(DateUtil.FORMAT_SS)+"MessageBroker=================================="+ ObjectToString.getString(message));
        messageQueue.enqueue(message);
    }

    public void removeListener(Object object){
        for(String queueName:queueMap.keySet()){
            MessageQueue messageQueue=queueMap.get(queueName);
            messageQueue.removeListener(object);
        }
    }
    public void removeListeners(List objects){
        for(Object object:objects){
            removeListener(object);
        }
    }
    public void addListener(String queueName, QueueListener queueListener){
        if(!queueMap.keySet().contains(queueName)){
            return;
        }
        MessageQueue messageQueue =queueMap.get(queueName);
        messageQueue.addListener(queueListener);
    }

    public void addListener(String queueName, QueueListener queueListener, boolean createQueueIfNotExist){
        if(!queueMap.keySet().contains(queueName) && !createQueueIfNotExist){
            return;
        }else if(!queueMap.keySet().contains(queueName) && createQueueIfNotExist){
            createQueue(queueName);
        }
        MessageQueue messageQueue =queueMap.get(queueName);
        messageQueue.addListener(queueListener);
    }
    public void createQueue(String queueName){
        queueMap.put(queueName,new AsyncMessageQueue(queueName));
    }
    public void removeQueue(String queueName){
        if(!queueMap.keySet().contains(queueName)){
            return;
        }
        MessageQueue queue=queueMap.get(queueName);
        queue.destroy();
        queueMap.remove(queueName);
    }

    /**
     * 没有监听者的队列
     * @param queueName
     * @return
     */
    public boolean isInactiveQueue(String queueName){
        if(existQueue(queueName)){
            return !queueMap.get(queueName).hasListener();
        }
        return false;
    }
    public boolean existQueue(String queueName){
        return queueMap.containsKey(queueName);
    }
}
